package com.atguigu.edu.realtime.common.util;

import com.atguigu.edu.realtime.common.constant.Constant;

public class SQLUtil {
    //kafka Connector
    public static String getKafkaDDL(String topic, String groupId) {
        return "   WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = '" + topic + "',\n" +
                "  'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "',\n" +
                "  'properties.group.id' = '" + groupId + "',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'format' = 'json'\n" +
                ")";
    }
    //获取HBase Connector
    public static String getHBaseDDL(String tableName){
        return "  WITH (\n" +
                " 'connector' = 'hbase-2.2',\n" +
                " 'table-name' = "+tableName+""+
                " 'zookeeper.quorum' = 'hadoop104,hadoop105,hadoop106:2181',\n" +
                " 'lookup.async' = 'true',\n" +
                " 'lookup.cache' = 'PARTIAL',\n" +
                " 'lookup.partial-cache.max-rows' = '500',\n" +
                " 'lookup.partial-cache.expire-after-write' = '1 day',\n" +
                " 'lookup.partial-cache.expire-after-access' = '1 day'\n" +
                ");";
    }

    //upsert kafka连接器
    public  static String getUpsertKafkaDDL(String topic){
        return "  WITH (\n" +
                "  'connector' = 'upsert-kafka',\n" +
                "  'topic' = '"+topic+"',\n" +
                "  'properties.bootstrap.servers' = 'hadoop104:9092,hadoop105:9092,hadoop106:9092',\n" +
                "  'key.format' = 'json',\n" +
                "  'value.format' = 'json'\n" +
                ");";
    }

    //doris connector
    public static String getDorisDDL(String tableName){
        return "     WITH (\n" +
                "    'connector' = 'doris',\n" +
                "    'fenodes' = '"+Constant.DORIS_FE_NODES+"',\n" +
                "    'table.identifier' = '"+Constant.DORIS_DATABASE+"."+tableName+"',\n" +
                "    'username' = 'root',\n" +
                "    'password' = 'aaaaaa',\n" +
                "    'sink.properties.format' = 'json',\n" +
                "    'sink.enable-2pc' = 'false'\n" +
                ")";
    }
}


